/*
 * Tencent is pleased to support the open source community by making BK-JOB蓝鲸智云作业平台 available.
 *
 * Copyright (C) 2021 Tencent.  All rights reserved.
 *
 * BK-JOB蓝鲸智云作业平台 is licensed under the MIT License.
 *
 * License for BK-JOB蓝鲸智云作业平台:
 * --------------------------------------------------------------------
 * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
 * documentation files (the "Software"), to deal in the Software without restriction, including without limitation
 * the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
 * to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
 * the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
 * THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
 * CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */

package com.tencent.bk.job.execute.engine.listener;

import com.google.common.collect.Lists;
import com.tencent.bk.job.common.constant.RollingModeEnum;
import com.tencent.bk.job.common.util.date.DateUtils;
import com.tencent.bk.job.execute.common.cache.CustomPasswordCache;
import com.tencent.bk.job.execute.common.constants.RunStatusEnum;
import com.tencent.bk.job.execute.common.util.TaskCostCalculator;
import com.tencent.bk.job.execute.engine.consts.JobActionEnum;
import com.tencent.bk.job.execute.engine.listener.event.JobEvent;
import com.tencent.bk.job.execute.engine.listener.event.JobMessage;
import com.tencent.bk.job.execute.engine.listener.event.StepEvent;
import com.tencent.bk.job.execute.engine.listener.event.TaskExecuteMQEventDispatcher;
import com.tencent.bk.job.execute.engine.model.JobCallbackDTO;
import com.tencent.bk.job.execute.engine.quota.limit.RunningJobResourceQuotaManager;
import com.tencent.bk.job.execute.model.RollingConfigDTO;
import com.tencent.bk.job.execute.model.StepInstanceBaseDTO;
import com.tencent.bk.job.execute.model.TaskInstanceDTO;
import com.tencent.bk.job.execute.model.db.ExecuteObjectRollingConfigDetailDO;
import com.tencent.bk.job.execute.service.NotifyService;
import com.tencent.bk.job.execute.service.rolling.RollingConfigService;
import com.tencent.bk.job.execute.service.StepInstanceService;
import com.tencent.bk.job.execute.service.TaskInstanceService;
import com.tencent.bk.job.execute.statistics.StatisticsService;
import com.tencent.bk.job.manage.GlobalAppScopeMappingService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.List;

/**
 * 执行引擎事件处理-作业
 */
@Component("jobEventListener")
@Slf4j
public class JobListener extends BaseJobMqListener {

    private final TaskExecuteMQEventDispatcher taskExecuteMQEventDispatcher;
    private final StatisticsService statisticsService;
    private final TaskInstanceService taskInstanceService;
    private final StepInstanceService stepInstanceService;
    private final RollingConfigService rollingConfigService;
    private final NotifyService notifyService;
    protected final CustomPasswordCache customPasswordCache;

    private final RunningJobResourceQuotaManager runningJobResourceQuotaManager;

    @Autowired
    public JobListener(TaskExecuteMQEventDispatcher taskExecuteMQEventDispatcher,
                       StatisticsService statisticsService,
                       TaskInstanceService taskInstanceService,
                       StepInstanceService stepInstanceService,
                       RollingConfigService rollingConfigService,
                       NotifyService notifyService,
                       RunningJobResourceQuotaManager runningJobResourceQuotaManager,
                       CustomPasswordCache customPasswordCache) {
        this.taskExecuteMQEventDispatcher = taskExecuteMQEventDispatcher;
        this.statisticsService = statisticsService;
        this.taskInstanceService = taskInstanceService;
        this.stepInstanceService = stepInstanceService;
        this.rollingConfigService = rollingConfigService;
        this.notifyService = notifyService;
        this.runningJobResourceQuotaManager = runningJobResourceQuotaManager;
        this.customPasswordCache = customPasswordCache;
    }


    /**
     * 处理作业执行相关的事件
     *
     * @param message 消息
     */
    @Override
    public void handleEvent(Message<? extends JobMessage> message) {
        JobEvent jobEvent = (JobEvent) message.getPayload();
        log.info("Handle job event, event: {}, duration: {}ms", jobEvent, jobEvent.duration());
        long jobInstanceId = jobEvent.getJobInstanceId();
        JobActionEnum action = JobActionEnum.valueOf(jobEvent.getAction());
        TaskInstanceDTO taskInstance;
        try {
            taskInstance = taskInstanceService.getTaskInstance(jobInstanceId);
            switch (action) {
                case START:
                    startJob(taskInstance);
                    break;
                case STOP:
                    stopJob(taskInstance);
                    break;
                case REFRESH:
                    refreshJob(taskInstance);
                    break;
                default:
                    log.error("Invalid job action: {}", action);
            }
        } catch (Throwable e) {
            String errorMsg = "Handle job event error, jobInstanceId=" + jobInstanceId;
            log.error(errorMsg, e);
        }
    }

    /**
     * 启动作业
     *
     * @param taskInstance 作业实例
     */
    private void startJob(TaskInstanceDTO taskInstance) {
        long jobInstanceId = taskInstance.getId();
        // 首先验证作业的状态，只有状态为“未执行”的作业可以启动
        if (RunStatusEnum.BLANK == taskInstance.getStatus()) {
            StepInstanceBaseDTO stepInstance = stepInstanceService.getFirstStepInstance(jobInstanceId);
            taskInstanceService.updateTaskExecutionInfo(jobInstanceId, RunStatusEnum.RUNNING, stepInstance.getId(),
                DateUtils.currentTimeMillis(), null, null);
            startStep(stepInstance);

            // 触发任务开始统计分析
            statisticsService.updateStartJobStatistics(taskInstance);
        } else {
            log.error("Unsupported job instance run status for starting job, jobInstanceId={}, status={}",
                jobInstanceId, taskInstance.getStatus());
        }
    }

    /**
     * 强制终止作业
     *
     * @param taskInstance 作业实例
     */
    private void stopJob(TaskInstanceDTO taskInstance) {
        long jobInstanceId = taskInstance.getId();
        RunStatusEnum taskStatus = taskInstance.getStatus();

        if (RunStatusEnum.RUNNING == taskStatus || RunStatusEnum.WAITING_USER == taskStatus) {
            taskInstanceService.updateTaskStatus(jobInstanceId, RunStatusEnum.STOPPING.getValue());
            long currentStepInstanceId = taskInstance.getCurrentStepInstanceId();
            taskExecuteMQEventDispatcher.dispatchStepEvent(
                StepEvent.stopStep(jobInstanceId, currentStepInstanceId));
        } else {
            log.warn("Unsupported job instance run status for stop task, jobInstanceId={}, status={}",
                jobInstanceId, taskInstance.getStatus());
        }
    }

    /**
     * 作业状态流转
     *
     * @param taskInstance 作业实例
     */
    private void refreshJob(TaskInstanceDTO taskInstance) {
        long jobInstanceId = taskInstance.getId();
        RunStatusEnum taskStatus = taskInstance.getStatus();

        long currentStepInstanceId = taskInstance.getCurrentStepInstanceId();
        StepInstanceBaseDTO currentStepInstance = stepInstanceService.getBaseStepInstance(
            jobInstanceId, currentStepInstanceId);
        RunStatusEnum stepStatus = currentStepInstance.getStatus();

        // 验证作业状态，只有'正在执行'、'强制终止中'的作业可以刷新状态进入下一步或者结束
        if (RunStatusEnum.STOPPING == taskStatus) {
            // 非正在执行的步骤可以直接终止
            if (RunStatusEnum.RUNNING != stepStatus) {
                finishJob(taskInstance, currentStepInstance, RunStatusEnum.STOP_SUCCESS);
            } else {
                log.error("Unsupported job instance run status for refresh task, jobInstanceId={}, taskStatus={}, " +
                    "stepStatus: {}", jobInstanceId, taskStatus, stepStatus);
            }
        } else if (RunStatusEnum.RUNNING == taskStatus) {
            // 步骤状态为成功、跳过、设为忽略错误、滚动等待，可以进入下一步
            if (RunStatusEnum.SUCCESS == stepStatus
                || RunStatusEnum.SKIPPED == stepStatus
                || RunStatusEnum.IGNORE_ERROR == stepStatus
                || RunStatusEnum.ROLLING_WAITING == stepStatus) {
                nextStep(taskInstance, currentStepInstance);
            } else if (RunStatusEnum.FAIL == stepStatus
                || RunStatusEnum.ABNORMAL_STATE == stepStatus
                || RunStatusEnum.ABANDONED == stepStatus) {
                // 步骤失败，任务结束
                finishJob(taskInstance, currentStepInstance, stepStatus);
            } else {
                log.warn("Unsupported job instance run status for refresh task, jobInstanceId={}, status={}",
                    jobInstanceId, taskInstance.getStatus());
            }
        } else {
            log.warn("Unsupported job instance run status for refresh task, jobInstanceId={}, status={}",
                jobInstanceId, taskInstance.getStatus());
        }
    }

    private void nextStep(TaskInstanceDTO taskInstance, StepInstanceBaseDTO currentStepInstance) {
        if (currentStepInstance.isRollingStep()) {
            RollingConfigDTO rollingConfig = rollingConfigService.getRollingConfig(
                currentStepInstance.getTaskInstanceId(),
                currentStepInstance.getRollingConfigId()
            );
            StepInstanceBaseDTO nextStepInstance = getNextStepInstance(
                taskInstance,
                currentStepInstance,
                rollingConfig
            );
            if (nextStepInstance == null) {
                finishJob(taskInstance, currentStepInstance, RunStatusEnum.SUCCESS);
            } else {
                RollingModeEnum mode = rollingConfig.getModeOfStep(currentStepInstance.getId());
                if (mode == RollingModeEnum.MANUAL
                    && rollingConfig.isFirstRollingStep(nextStepInstance.getId())) {
                    log.info("Manual mode for rolling step[{}], pause and wait for user confirmation",
                        nextStepInstance.getId());
                    stepInstanceService.updateStepStatus(taskInstance.getId(), nextStepInstance.getId(),
                        RunStatusEnum.WAITING_USER.getValue());
                    taskInstanceService.updateTaskStatus(taskInstance.getId(), RunStatusEnum.WAITING_USER.getValue());
                } else {
                    // 执行下一步骤
                    startStep(nextStepInstance);
                }
            }
        } else {
            StepInstanceBaseDTO nextStepInstance = getNextStepInstance(taskInstance, currentStepInstance, null);
            if (nextStepInstance == null) {
                finishJob(taskInstance, currentStepInstance, RunStatusEnum.SUCCESS);
            } else {
                // 执行下一步骤
                startStep(nextStepInstance);
            }
        }

    }

    private void finishJob(TaskInstanceDTO taskInstance,
                           StepInstanceBaseDTO stepInstance,
                           RunStatusEnum jobStatus) {
        long jobInstanceId = taskInstance.getId();
        try {
            Long endTime = DateUtils.currentTimeMillis();
            long totalTime = TaskCostCalculator.calculate(taskInstance.getStartTime(), endTime,
                taskInstance.getTotalTime());
            taskInstance.setEndTime(endTime);
            taskInstance.setTotalTime(totalTime);
            taskInstance.setStatus(jobStatus);
            taskInstanceService.updateTaskExecutionInfo(jobInstanceId, jobStatus, null, null, endTime, totalTime);

            // 处理目标主机的密码缓存
            handleTargetHostPasswordCache(jobInstanceId, jobStatus);

            // 作业执行结果消息通知
            if (RunStatusEnum.SUCCESS == jobStatus || RunStatusEnum.IGNORE_ERROR == jobStatus) {
                notifyService.asyncSendMQSuccessTaskNotification(taskInstance, stepInstance);
            } else {
                notifyService.asyncSendMQFailTaskNotification(taskInstance, stepInstance);
            }

            // 触发作业结束统计分析
            statisticsService.updateEndJobStatistics(taskInstance);

            // 作业执行完成回调
            callback(taskInstance);
        } finally {
            // 从资源配额中删除该作业实例
            runningJobResourceQuotaManager.removeJob(
                taskInstance.getAppCode(),
                GlobalAppScopeMappingService.get().getScopeByAppId(taskInstance.getAppId()),
                jobInstanceId
            );
        }

    }

    private void handleTargetHostPasswordCache(Long jobInstanceId, RunStatusEnum jobStatus) {
        if (RunStatusEnum.SUCCESS == jobStatus || RunStatusEnum.IGNORE_ERROR == jobStatus) {
            customPasswordCache.deleteCache(jobInstanceId);
        } else {
            customPasswordCache.setPwdExpireTimeOnTaskFail(jobInstanceId);
        }
    }

    /**
     * 获取下一个执行的步骤实例
     *
     * @param taskInstance        作业实例
     * @param currentStepInstance 当前步骤实例
     * @param rollingConfigDTO       滚动配置，仅当当前步骤为滚动步骤时才需要传入
     * @return 步骤实例;如果当前步骤已经是最后一个步骤，那么返回null
     */
    private StepInstanceBaseDTO getNextStepInstance(TaskInstanceDTO taskInstance,
                                                    StepInstanceBaseDTO currentStepInstance,
                                                    RollingConfigDTO rollingConfigDTO) {
        StepInstanceBaseDTO nextStepInstance;
        if (currentStepInstance.isRollingStep()) {
            if (rollingConfigDTO.isFileSourceRolling()) {
                // 按源文件滚动的步骤
                nextStepInstance = getNextStepInstanceForFileSourceRollingStep(
                    taskInstance,
                    currentStepInstance,
                    rollingConfigDTO
                );
            } else {
                // 按目标执行对象滚动的步骤
                nextStepInstance = getNextStepInstanceForEORollingStep(
                    taskInstance,
                    currentStepInstance,
                    rollingConfigDTO
                );
            }
        } else {
            nextStepInstance = stepInstanceService.getNextStepInstance(taskInstance.getId(),
                currentStepInstance.getStepOrder());
        }
        return nextStepInstance;
    }

    /**
     * 当前步骤为按源文件滚动的步骤，获取下一个步骤
     *
     * @param taskInstance        任务实例
     * @param currentStepInstance 当前步骤实例
     * @param rollingConfigDTO    任务级别的滚动配置
     * @return 下一个步骤
     */
    private StepInstanceBaseDTO getNextStepInstanceForFileSourceRollingStep(TaskInstanceDTO taskInstance,
                                                                            StepInstanceBaseDTO currentStepInstance,
                                                                            RollingConfigDTO rollingConfigDTO) {
        StepInstanceBaseDTO nextStepInstance;
        int totalBatch = rollingConfigService.getTotalBatch(
            taskInstance.getId(),
            currentStepInstance.getId(),
            rollingConfigDTO.getId()
        );
        int currentBatch = currentStepInstance.getBatch();
        boolean isLastBatch = totalBatch == currentBatch;

        if (isLastBatch) {
            // 最后一个滚动批次，那么该滚动任务执行结束，进入下一个步骤
            nextStepInstance = stepInstanceService.getNextStepInstance(
                taskInstance.getId(),
                currentStepInstance.getStepOrder()
            );
        } else {
            // 非最后一个批次，那么继续执行当前步骤的下一个批次
            nextStepInstance = currentStepInstance;
        }
        return nextStepInstance;
    }

    /**
     * 当前步骤为按目标执行对象滚动的步骤，获取下一个步骤
     *
     * @param taskInstance              任务实例
     * @param currentStepInstance       当前步骤实例
     * @param taskInstanceRollingConfig 任务级别的滚动配置
     * @return 下一个步骤
     */
    private StepInstanceBaseDTO getNextStepInstanceForEORollingStep(TaskInstanceDTO taskInstance,
                                                                    StepInstanceBaseDTO currentStepInstance,
                                                                    RollingConfigDTO taskInstanceRollingConfig){
        StepInstanceBaseDTO nextStepInstance = null;
        ExecuteObjectRollingConfigDetailDO rollingConfig = taskInstanceRollingConfig.getExecuteObjectRollingConfig();
        int currentBatch = currentStepInstance.getBatch();
        List<Long> includeStepInstanceIdList = rollingConfig.getIncludeStepInstanceIdList();
        boolean isLastRollingStep = rollingConfig.isLastRollingStep(currentStepInstance.getId());
        boolean isLastBatch = rollingConfig.getTotalBatch() == currentBatch;

        // 最后一个滚动步骤和滚动批次，那么该滚动任务执行结束，进入下一个步骤
        if (isLastRollingStep && isLastBatch) {
            nextStepInstance = stepInstanceService.getNextStepInstance(taskInstance.getId(),
                currentStepInstance.getStepOrder());
        } else {
            Long nextRollingStepInstanceId;
            if (isLastRollingStep) {
                nextRollingStepInstanceId = includeStepInstanceIdList.get(0);
            } else {
                nextRollingStepInstanceId = getNextStepInstanceId(includeStepInstanceIdList,
                    currentStepInstance.getId());
            }
            if (nextRollingStepInstanceId != null) {
                nextStepInstance = stepInstanceService.getBaseStepInstance(taskInstance.getId(),
                    nextRollingStepInstanceId);
            }
        }
        return nextStepInstance;
    }

    private Long getNextStepInstanceId(List<Long> stepInstanceIdList, long currentStepInstanceId) {
        int currentStepIndex = stepInstanceIdList.indexOf(currentStepInstanceId);
        // 当前步骤为最后一个步骤
        if (currentStepIndex == stepInstanceIdList.size() - 1) {
            return null;
        }
        return stepInstanceIdList.get(currentStepIndex + 1);
    }

    private void startStep(StepInstanceBaseDTO stepInstance) {
        taskInstanceService.updateTaskCurrentStepId(stepInstance.getTaskInstanceId(), stepInstance.getId());
        if (stepInstance.isRollingStep()) {
            stepInstanceService.updateStepStatus(
                stepInstance.getTaskInstanceId(),
                stepInstance.getId(),
                RunStatusEnum.ROLLING_WAITING.getValue()
            );
            taskExecuteMQEventDispatcher.dispatchStepEvent(
                StepEvent.startStep(
                    stepInstance.getTaskInstanceId(),
                    stepInstance.getId(),
                    stepInstance.getBatch() + 1
                )
            );
        } else {
            stepInstanceService.updateStepStatus(
                stepInstance.getTaskInstanceId(),
                stepInstance.getId(),
                RunStatusEnum.BLANK.getValue()
            );
            taskExecuteMQEventDispatcher.dispatchStepEvent(
                StepEvent.startStep(
                    stepInstance.getTaskInstanceId(),
                    stepInstance.getId(),
                    null
                )
            );
        }
    }

    private void callback(TaskInstanceDTO taskInstance) {
        if (StringUtils.isNotBlank(taskInstance.getCallbackUrl())) {
            JobCallbackDTO callback = new JobCallbackDTO();
            callback.setId(taskInstance.getId());
            callback.setStatus(taskInstance.getStatus().getValue());
            callback.setCallbackUrl(taskInstance.getCallbackUrl());
            Collection<JobCallbackDTO.StepInstanceStatus> stepInstanceStatusList = Lists.newArrayList();
            List<StepInstanceBaseDTO> stepInstanceList =
                stepInstanceService.listBaseStepInstanceByTaskInstanceId(taskInstance.getId());
            stepInstanceList.forEach(stepInstance -> {
                JobCallbackDTO.StepInstanceStatus stepInstanceStatus = new JobCallbackDTO.StepInstanceStatus();
                stepInstanceStatus.setId(stepInstance.getId());
                stepInstanceStatus.setStatus(stepInstance.getStatus().getValue());
                stepInstanceStatusList.add(stepInstanceStatus);
            });
            callback.setStepInstances(stepInstanceStatusList);
            taskExecuteMQEventDispatcher.dispatchCallbackMsg(callback);
        }
    }

}
